package com.fulu.game.bigdata.realtime.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.fulu.game.bigdata.realtime.config.{Config, Constants}
import com.fulu.game.bigdata.realtime.entity.Sink
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.types.Row
import org.slf4j.LoggerFactory

import scala.collection.JavaConverters._

class MySqlSink(sink : Sink) extends RichSinkFunction[(Boolean,Row)] {

  private val logger = LoggerFactory.getLogger(classOf[MySqlSink])

  private var connection: Connection = null
  private var preparedStatement: PreparedStatement = null

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    Class.forName(Constants.DB_DRIVER)
    val url = Config.getMySqlSinkUrl()
    connection = DriverManager.getConnection(url, Config.getMySqlSinkUsername(), Config.getMySqlSinkPassword())
    val sql = generateSql(sink)
    preparedStatement = connection.prepareStatement(sql)
  }

  override def invoke(value: (Boolean, Row), context: SinkFunction.Context[_]): Unit = {
    if(!value._1) return
    val result = value._2
    val fieldSize = sink.getFields.size()
    for( i <- 0 to fieldSize-1 ){
      preparedStatement.setObject( i+1 , result.getField(i))
      println((i+1) + " --- " + result.getField(i) )
    }
    preparedStatement.executeUpdate()
  }

  override def close(): Unit = {
    if (preparedStatement != null) {
      preparedStatement.close()
    }
    if (connection != null) {
      connection.close()
    }
  }

  private def generateSql(sink : Sink): String ={

    val sb = new StringBuffer

    if (sink.getModel == "update" ){
      sb.append("update ").append(sink.getSinkTable).append(" set ")

      var updateFields = sink.fields.asScala.filter(field => !field.isPrimaryKey && !field.isUniqueKey).map(_.getName + "=?").mkString(",")
      if(updateFields.endsWith(",")){
        updateFields = updateFields.substring(0,updateFields.length-1)
      }

      sb.append(updateFields).append(" where ")

      var conditionFields = sink.fields.asScala.filter(field => field.isPrimaryKey || field.isUniqueKey).map(_.getName + "=?").mkString(" and ").trim
      if(conditionFields.endsWith("and")){
        conditionFields = conditionFields.substring(0,conditionFields.length -3 )
      }

      sb.append(conditionFields)
    } else if(sink.getModel == "insert"){

      sb.append("insert into ").append(sink.getSinkTable).append(" (")
      var fields = sink.fields.asScala.map(_.getName).mkString(",")
      if(fields.endsWith(",")){
        fields = fields.substring(0,fields.length -1)
      }
      sb.append(fields).append(") values (")

      val placeholders = sink.getFields.asScala.map(fields => "?").mkString(",")
      sb.append(placeholders).append(")")
    }
    println(sb.toString)
    sb.toString
  }
}
